// Project name:JavaRedis
// Creator:muhong
// Date time:2022/12/8,9:36 下午
// Name:AbstractChannel

package org.gmh.channel;

import org.gmh.channel.handler.ChannelPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;

public abstract class AbstractChannel implements Channel {

    private final static Logger logger = LoggerFactory.getLogger(AbstractChannel.class);

    private final Channel parent;
    private final Unsafe unSafe;
    private final DefaultChannelPipeline pipeline;
    private final CloseFuture closeFuture = new CloseFuture(this);
    private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);

    private volatile EventLoop eventLoop;
    private volatile boolean registered;
    private volatile SocketAddress localAddress;
    private volatile SocketAddress remoteAddress;
    private boolean closeInitiated;


    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        this.unSafe = newUnsafe();
        this.pipeline = newChannelPipeline();
    }

    protected abstract AbstractUnsafe newUnsafe();

    protected abstract DefaultChannelPipeline newChannelPipeline();

    @Override
    public Unsafe unsafe() {
        return unSafe;
    }

    protected void doRegister() throws Exception {

    }

    @Override
    public EventLoop eventLoop() {
        return this.eventLoop;
    }

    @Override
    public boolean isRegistered() {
        return registered;
    }

    @Override
    public ChannelPipeline pipeline() {
        return this.pipeline;
    }

    protected abstract void doWrite(Object msg) throws Exception;

    protected abstract void doClose() throws Exception;

    protected void doDeregister() throws Exception {

    }

    @Override
    public Channel parent() {
        return parent;
    }

    @Override
    public ChannelFuture closeFuture() {
        return closeFuture;
    }

    @Override
    public ChannelFuture bind(SocketAddress localAddress) {
        return null;
    }

    @Override
    public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture disconnect(ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) {
        return null;
    }

    @Override
    public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture connect(SocketAddress remoteAddress) {
        return null;
    }

    @Override
    public ChannelFuture deregister(ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture deregister() {
        return null;
    }

    @Override
    public ChannelFuture close(ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture close() {
        return null;
    }

    @Override
    public ChannelFuture write(Object msg) {
        return null;
    }

    @Override
    public ChannelFuture write(Object msg, ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
        return null;
    }

    @Override
    public ChannelFuture writeAndFlush(Object msg) {
        return null;
    }

    @Override
    public ChannelProgressivePromise newProgressivePromise() {
        return null;
    }

    @Override
    public ChannelPromise newPromise() {
        return pipeline.newPromise();
    }

    @Override
    public ChannelFuture newFailedFuture() {
        return null;
    }

    @Override
    public ChannelFuture newSucceededFuture() {
        return null;
    }

    @Override
    public ChannelPromise voidPromise() {

        return pipeline.voidPromise();
    }

    @Override
    public ChannelFuture disconnect() {
        return null;
    }

    @Override
    public Channel read() {
        return null;
    }

    @Override
    public Channel flush() {
        return null;
    }

    protected abstract class AbstractUnsafe implements Unsafe {

        private boolean neverRegistered = true;

        @Override
        public void bind(SocketAddress socketAddress) {

        }

        @Override
        public void register(EventLoop eventLoop, ChannelPromise promise) {
            // 执行注册
            AbstractChannel.this.eventLoop = eventLoop;
            try {
                if (eventLoop.inEventLoop()) {
                    logger.info("the thread is in event loop");
                    register0(promise);
                } else {
                    // 这里初次初始化，以任务的方式加入到loop的任务队列中
                    eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                register0(promise);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    });
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        private void register0(ChannelPromise promise) {
            try {
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
                boolean firstRegistration = neverRegistered;
                doRegister();
                neverRegistered = false;
                registered = true;

                pipeline.invokeHandlerAddedIfNeeded();

                safeSetSuccess(promise);
                pipeline.fireChannelRegistered();
                if (isActive()) {
                    if (firstRegistration) {
                        pipeline.fireChannelActive();
                    }
                }
            } catch (Throwable t) {
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }

        @Override
        public void write(Object msg) {
            try {
                doWrite(msg);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public final void close(ChannelPromise promise) {
            assertEventLoop();

            close(promise, null);
        }

        @Override
        public ChannelPromise voidPromise() {
            assertEventLoop();
            return unsafeVoidPromise;
        }

        @Override
        public void closeForcibly() {
            assertEventLoop();
            try {
                doClose();
            } catch (Exception e) {
                logger.warn("Failed to close a channel.", e);
            }
        }

        private void close(final ChannelPromise promise, final Throwable cause) {
            if (!promise.setUncancellable()) {
                return;
            }
            if (closeInitiated) {
                if (closeFuture.isDone()) {
                    safeSetSuccess(promise);
                } else if (!(promise instanceof VoidChannelPromise)) {
                    closeFuture.addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            promise.setSuccess();
                        }
                    });
                }
                return;
            }

            closeInitiated = true;
            final boolean wasActive = isActive();
            // 判断是否在event loop中
            try {
                doClose0(promise);
                //
                closeFuture.setClosed();
                //
                fireChannelInactiveAndDeRegister();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        private void doClose0(ChannelPromise promise) {
            try {
                doClose();
                closeFuture.setClosed();
                safeSetSuccess(promise);
            } catch (Throwable t) {
                closeFuture.setFailure(t);
                safeSetFailure(promise, t);
            }
        }

        private void fireChannelInactiveAndDeRegister() {
            deregister();
        }

        private void deregister() {
            try {
                doDeregister();
            } catch (Throwable t) {
                logger.warn("Unexpected exception occurred while deregistering a channel.", t);
            } finally {
                pipeline.fireChannelUnregistered();
            }
        }

        private void assertEventLoop() {
            assert !registered || eventLoop.inEventLoop();
        }


        protected final void safeSetSuccess(ChannelPromise promise) {
            if ((!(promise instanceof VoidChannelPromise)) && !promise.trySuccess()) {
                logger.warn("Failed to mark a promise as success because it is done already:{}", promise);
            }
        }

        protected final void safeSetFailure(ChannelPromise promise, Throwable cause) {
            if ((!(promise instanceof VoidChannelPromise)) && !promise.tryFailure(cause)) {
                logger.warn("Failed to mark a promise as success because it is done already:{}", promise);
            }
        }

        protected final boolean ensureOpen(ChannelPromise promise) {
            return true;
        }
    }

    static final class CloseFuture extends DefaultChannelPromise {

        public CloseFuture(Channel channel) {
            super(channel);
        }

        @Override
        public ChannelPromise setSuccess(Void result) {
            throw new IllegalStateException();
        }

        @Override
        public ChannelPromise setFailure(Throwable cause) {
            throw new IllegalStateException();
        }

        @Override
        public boolean trySuccess() {
            throw new IllegalStateException();
        }

        @Override
        public boolean tryFailure(Throwable cause) {
            throw new IllegalStateException();
        }

        boolean setClosed() {
            return super.trySuccess();
        }
    }
}
